GROOVY-12459: Failed AsyncChannel producer never closes the channel so for await hangs and the exception is dropped - #2987

JMH summary — classic (commit 2ab1cc7)

Speedup vs trailing 90-day baseline on gh-pages. Higher = faster.
1.00 = in line with history. Per-benchmark ratio, geomean within group.
Time-per-op units inverted so direction is consistent. The calibrated
column divides out this runner's speed vs the baseline hardware, as
measured by Groovy-independent pure-Java ruler benchmarks.

No benchmark is ≥1.5× slower than its 90-day baseline.

⚠️ Runner speed differs ≥15% from the historical baseline hardware for: bench, core-hz, grails-ad. Raw speedups are not meaningful for those parts — use the calibrated column.

_{Runner calibration (this run vs baseline hardware): bench 1.69× (27 rulers) · core-ag 0.97× (3 rulers) · core-hz 1.29× (3 rulers) · grails-ad 1.21× (3 rulers) · grails-ez 0.92× (3 rulers)}

_{Baseline: dev/bench/jmh/<part>/classic/data.js on gh-pages, trailing 90 days. Daily dashboard · Per-suite raw data}

JMH summary — indy (commit 2ab1cc7)

Speedup vs trailing 90-day baseline on gh-pages. Higher = faster.
1.00 = in line with history. Per-benchmark ratio, geomean within group.
Time-per-op units inverted so direction is consistent. The calibrated
column divides out this runner's speed vs the baseline hardware, as
measured by Groovy-independent pure-Java ruler benchmarks.

⚠️ 2 benchmarks at least 1.5× slower than the 90-day baseline:

• org.apache.groovy.bench.AryBench.groovyCS ( {"n":"1000000"} ) — 2.03× slower (calibrated)
• org.apache.groovy.bench.StaticMethodCallIndyBench.staticChain_groovyCS — 1.55× slower (calibrated)

⚠️ Runner speed differs ≥15% from the historical baseline hardware for: core-hz. Raw speedups are not meaningful for those parts — use the calibrated column.

_{Runner calibration (this run vs baseline hardware): bench 1.01× (27 rulers) · core-ag 0.95× (3 rulers) · core-hz 0.85× (3 rulers) · grails-ad 0.90× (3 rulers) · grails-ez 0.88× (3 rulers)}

_{Baseline: dev/bench/jmh/<part>/indy/data.js on gh-pages, trailing 90 days. Daily dashboard · Per-suite raw data}

Codecov Report

❌ Patch coverage is 91.86047% with 14 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.4638%. Comparing base (fa53afa) to head (ed5ed29).
⚠️ Report is 1 commits behind head on master.

@@                Coverage Diff                 @@
##               master      #2987        +/-   ##
==================================================
+ Coverage     72.4452%   72.4638%   +0.0186%     
- Complexity      40687      40721        +34     
==================================================
  Files            1644       1644                
  Lines          142897     143008       +111     
  Branches        26877      26895        +18     
==================================================
+ Hits           103522     103629       +107     
- Misses          29820      29826         +6     
+ Partials         9555       9553         -2     

Current solution borrows ideas from C#, it looks not very groovy, we can try to find better solution.

using System.Threading.Channels;

var channel = Channel.CreateBounded<int>(2);

_ = Task.Run(async () =>
{
    try
    {
        for (var i = 1; i <= 3; i++)
        {
            if (i == 3)
                throw new InvalidOperationException("sensor failed");
            await channel.Writer.WriteAsync(i * i);
        }
        channel.Writer.Complete();
    }
    catch (Exception failure)
    {
        channel.Writer.Complete(failure);
    }
});

var seen = new List<int>();
try
{
    await foreach (var value in channel.Reader.ReadAllAsync())
        seen.Add(value);
}
catch (InvalidOperationException e)
{
    Console.WriteLine(e.Message); // sensor failed
}

// seen == [1, 4]

Here is the JavaScript solution:

let n = 0;
const stream = new ReadableStream({
  pull(controller) {
    n += 1;
    if (n === 3) {
      controller.error(new Error('sensor failed')); // close with the cause
      return;
    }
    controller.enqueue(n * n);
  }
});

const seen = [];
try {
  for await (const value of stream) {
    seen.push(value);
  }
} catch (failure) {
  console.log(failure.message); // sensor failed
}

AI read:

PR #2987 fixes a real problem and the approach is sound, but it needs a few changes before merging. I'd ask for three changes (listed below) and also suggest a producer helper that closes the channel automatically. It's from daniellansun, one commit 570c9124bc, +1019/−107.

What it does

Until now, a producer that threw before calling close() left for await waiting forever, and the exception was lost. The PR adds close(Throwable) and getCloseCause() to AsyncChannel, both as default methods. DefaultAsyncChannel keeps the failure and attaches it as the cause of ChannelClosedException. Its iterator rethrows the failure itself, which is what makes for await throw. map, filter, merge, split and tap now pass a failure through, ChannelSelect fails with it, and BroadcastChannel gets close(Throwable) and reports it to subscribers as onError.

Testing: CI ran 33/33 checks with 122,256 tests passing, but the SonarCloud gate failed on "Reliability Rating on New Code". The Sonar lookup was blocked here, so I don't know which issue it is. Locally on JDK 17, the PR's own tests plus some probe tests I wrote all pass.

What's good

• The model has precedent. Kotlin's Channel.close(cause) works the same way: values already sent still arrive, then the failure.
• Old custom channels can't hang. The default close(Throwable) just calls close(), so a custom implementation built against 6.0.0 still ends the loop.
• Composition is cleaner. One shared pump/offer helper replaces five copies of the receive loop. A callback that throws inside map or filter is no longer lost on an executor thread.
• ChannelSelect is careful. A failure can't override a transfer that has already committed.
• It also fixes an older race. BroadcastChannel.subscribe used to race with close; it now takes the same lock.

Changes I'd request

1. Make close(Throwable) return boolean. close() returns boolean but the new method returns void (probe confirmed), while Kotlin's close(cause) returns Boolean. This is easy to change now and impossible after release.
2. merge holds the failure back. When one input fails and the other is still open, the merged channel stays open with no cause. The failure only shows up once the other input closes, so a long-running second input hides it indefinitely. The merged output should close with the failure straight away.
3. The Javadoc describes history. For example: "so a channel compiled against 6.0.0 does not hang". It should describe current behaviour, and the history belongs in the commit message.

Design points to raise

• Hand-written receive loops still drop the failure. receive and send throw ChannelClosedException with the failure as its cause, while iteration throws the failure itself. Many people end a manual loop with catch (ChannelClosedException), which is exactly what the library's own operators did before this PR. Such loops end quietly and lose the failure (probe confirmed). Kotlin avoids this by throwing the cause directly from receive. If this design stays, the docs should tell people writing manual loops to check getCause().
• More than for await is affected. for await is just the channel's iterator, so plain for, each, collect and toList on a failed channel now throw too (probe: each threw IOException after [1, 2]). Java for-each loops get an undeclared checked exception from hasNext(), which Java code can't catch by its specific type. This deserves a sentence in the docs.
• The producer side is still mostly boilerplate. Every producer has to write try / close / catch / close(failure) itself, which is roughly the trap the issue describes. I'd suggest a helper like Kotlin's produce { }: it runs the closure asynchronously and closes the channel normally or with the failure. close(Throwable) would remain the underlying method. This is probably the "groovier" answer Daniel is looking for, and it could be a follow-up.
• Select treats the two kinds of close differently. A channel closed normally is skipped until every channel is closed, but a channel closed with a failure fails the whole select. Which one wins against a ready value depends only on the order the channels are listed (50/50 when I alternated the order). That's consistent and matches Kotlin; it's just worth knowing.

Minor

• Docs style. The user guide names C#, JavaScript and Swift, which is unusual for the Groovy docs. "A channel compiled against the previous release" means a custom AsyncChannel implementation and should say so.
• Bundled fix. The BroadcastChannel subscribe/close race fix is a separate bug fix inside this PR. That's fine, but the commit message should mention it.

@paulk-asert Thank you for the careful read, and for checking the behaviour locally. The three changes you asked for are in, and the documentation points are covered as well.

close(Throwable) returns boolean

AsyncChannel.close(Throwable) now returns boolean, the same way close() does. true means this call closed the channel. The first close wins, so a later call returns false and leaves the retained failure unchanged. null is a normal close.

The default method calls close() and returns that result. It does not retain the cause, so an implementation that does not override the method still ends iteration instead of throwing. DefaultAsyncChannel retains the cause and returns the same boolean from finishClose.

BroadcastChannel.close(Throwable) returns the same boolean. BroadcastChannel.close() stays void, because that method shipped in 6.0.0.

merge closes on the first failure

When one input fails, the merged channel closes with that failure. It does not stay open until the other input ends. Values a relay has already taken, and values already buffered on the other input, are forwarded first. The other input is left open and is not read further. A producer that has not yet handed a value over is not released. When both inputs fail, the first failure retained on the output wins, and a later normal close does not replace it.

Javadoc and the user guide

The Javadoc now describes the current contract. The note about a channel compiled against 6.0.0 is gone; that history belongs in the commit message. The guide says "a custom AsyncChannel that does not override close(Throwable)".

The guide also says:

• a receive / send loop must read ChannelClosedException.getCause(), or the failure is dropped
• plain for, each, collect, and toList use the same iterator and throw the failure
• a Java enhanced for calls Iterator.hasNext(), which can throw a checked exception that Iterator does not declare
• a normal close and a failure close are different in select: a buffered value is still delivered, and the failure is the result only when nothing buffered remains. Which one is met first follows registration order (list order, fair(), or random())

The failure section no longer names C#, JavaScript, or Swift.

Follow-ups

A produce { } helper would remove the try / close / catch boilerplate, with close(Throwable) still the primitive underneath. I agree that is the groovier next step, and I have left it as a follow-up so this change stays the contract fix.

The BroadcastChannel subscribe / close race is a separate fix inside the same commit. I will name it in the commit message when this lands. Suggested sentence: "Also close the race between BroadcastChannel.subscribe and close."

Sonar

volatile Throwable closeCause is the reference publication Sonar flags as java:S3077, which is what failed the reliability rating on new code. The field is marked the same way as the other immutable holders in the tree. The Throwable is not mutated after it is published.

Looks good to me overall. Close-with-cause is opt-in, so a plain close() behaves exactly as before, and it reaches iteration, the filter/map/split/tap pipelines, merge, the Reactive Streams bridge (onError) and select.

Checked locally, merged with current master:

• groovy.concurrent.*, *Async* and *Channel* tests: 1846 run, 0 failures.
• AsyncChannelFailureTest, BroadcastChannelTest and BroadcastChannelAsPublisherTest: 8 forced reruns, no flakes.

One thing to fix before merging: a value can be lost in merge. In AsyncChannel.relay (src/main/java/groovy/concurrent/AsyncChannel.java, around line 391):

if (self.isStopping() && !pending.isDone()) {
    pending.cancel();
    break;
}

A sender can commit its transfer to pending between the isDone() check and cancel(). The claim then refuses the cancel (cancel() returns false), but we break anyway, so a value already removed from the input is dropped. The sender's send has completed successfully, so nobody sees an error. That contradicts the merge Javadoc ("every value a relay has already taken … has been forwarded").

Suggested fix: only break when the cancel actually wins, and otherwise fall through to the await so the value is forwarded:

if (self.isStopping() && !pending.isDone() && pending.cancel()) {
    break;
}

The next iteration then exits through the isStopping() && getBufferedSize() == 0 check as usual. It's hard to hit in a test, but it's a one-line change.

Not blocking, just noting: when one input fails, merge keeps forwarding the other input's buffered values, so closing the output can wait a while behind a fast producer on that input. The docs describe this ("every value already buffered on the other input").

@paulk-asert Thank you. The local runs are reassuring, and the lost-value race is fixed.

Cancel that loses the commit

relay now breaks only when cancel() actually wins:

if (self.isStopping() && !pending.isDone() && pending.cancel()) {
    break;
}

If a sender commits between isDone() and cancel(), cancel() returns false and the receive falls through to await, so the value is forwarded. The next iteration still stops through isStopping() && getBufferedSize() == 0. A deterministic stand-in covers that lost cancel: isDone() stays false and cancel() completes the value but returns false.

Buffered values on the other input

Left as specified. While the other input's buffer stays non-empty, merge keeps forwarding, so a producer that refills the buffer can delay the failure close until a gap appears. The Javadoc and the user guide say so.

produce

AsyncChannel.produce and produce(capacity) run the body on the async executor and close the channel when it finishes. Producer.send waits until the channel accepts the value, including a rendezvous. A normal return closes the channel normally. A thrown failure is passed to close(cause), after the values already accepted. A ChannelClosedException from send after the consumer has already closed the channel is not recorded as a new failure. close(Throwable) stays the primitive underneath.

def channel = AsyncChannel.produce(2) {
    it.send(1)
    it.send(4)
}

Quality Gate passed

Issues
9 New issues
0 Accepted issues

Measures
0 Security Hotspots
93.3% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud